package hyl.base.mq.rabbit;

import java.io.IOException;
import java.net.URISyntaxException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.MessageProperties;

import hyl.base.mq.BaseMQ;
import hyl.core.Amy;
import hyl.core.MyFun;
import hyl.core.conf.MyConfig;
import hyl.core.info.CMsg;
import hyl.core.run.IFinishT;

/** 简易版本的rabbit消息组件,足够用了 */
public class MyRabbitMq extends BaseMQ {
	/** 全网广播不适合消息中间件 */
	// static final int M_广播=1;
	public static final int M_一对一 = 2;
	/** 一对多,排队轮流谈 */
	public static final int M_轮发 = 3;
	/** 一对多组播 */
	public static final int M_群发 = 4;
	/** 一对多组播-多个特定主题 */
	public static final int M_群发精确主题 = 5;
	/** 一对多组播 相似主题 */
	public static final int M_群发模糊主题 = 6;
	static final String file_conf = "mq/rabbitmq.conf";
	public static final Logger Log = LoggerFactory.getLogger(MyRabbitMq.class);
	Connection connection;
	Channel _channel;
	int _方式 = 0;
	boolean _is自动应答 = true;

	private MyRabbitMq(int 传送方式) {
		_方式 = 传送方式;
	}

	/**
	 * 轮发模式下 应答方式必须强制为false
	 * 
	 * @param 应答方式
	 * @return
	 */

	public boolean is自动应答() {
		return _is自动应答;
	}

	public static MyRabbitMq getInstance(int 传送方式) {
		return getInstance(传送方式, null);
	}

	public static MyRabbitMq getInstance(int 传送方式, String configfile) {
		MyRabbitMq mrm = new MyRabbitMq(传送方式);
		try {
			return mrm.setConnectConfig(configfile);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	public MyRabbitMq setConnectConfig(String file) throws Exception {
		Map<String, String> rbmqProperty;
		if (!MyFun.isEmpty(file))
			rbmqProperty = MyConfig.loadIni(file,  Amy.charset);
		else
			rbmqProperty = MyConfig.loadIni(file_conf,  Amy.charset);
		String username = rbmqProperty.get("USERNAME");
		String password = rbmqProperty.get("PASSWORD");
		String host = rbmqProperty.get("HOST");
		String url = String.format("amqp://%s:%s@%s", username, password, host);
		connect(url);
		return this;
	}

	public void connect(String url)
			throws KeyManagementException, NoSuchAlgorithmException, URISyntaxException, IOException, TimeoutException {
		/**
		 * 创建连接连接到MabbitMQ
		 */
		ConnectionFactory factory = new ConnectionFactory();
		// 设置MabbitMQ所在主机ip或者主机名
		// factory.setHost("192.168.0.144");
		// factory.setPort(5672);
		// 原guest只能本地连接 ,远程连接必须再创建一个admin
		// System.out.println(url);
		factory.setUri(url);
		// 创建一个连接
		connection = factory.newConnection();
		// 创建一个频道
		_channel = connection.createChannel();
	}

	void onBeforeAction() throws IOException {
		switch (_方式) {
		case M_一对一:
			_channel.queueDeclare(_key, false, false, false, null);
			break;
		case M_轮发:
			_channel.queueDeclare(_key, true, false, false, null);
			break;
		case M_群发:
			_channel.exchangeDeclare(_topic, "fanout");
			break;
		case M_群发精确主题:
			_channel.exchangeDeclare(_topic, "direct");
			break;
		case M_群发模糊主题:
			_channel.exchangeDeclare(_topic, "topic");
			break;
		}
	}

	@Override
	/**
	 * 如果为空 ,用ready ()的默认topic
	 */
	public void send(String topic,byte[] msg) {
		if (msg == null)
			return;
		if (MyFun.isEmpty(_topic))
			return;
		if(!MyFun.isEmpty(topic))
			_key=topic;
		if(!MyFun.isEmpty(_key))
			return;
		CMsg m = new CMsg();
		m.s主题 = _key;
		m.bs内容 = msg;
		try {
			onBeforeAction();
			switch (_方式) {
			case M_一对一:
				_channel.basicPublish("", _key, null, msg);
				break;
			case M_轮发:
				_channel.basicPublish("", _key, MessageProperties.PERSISTENT_BASIC, msg);
				break;
			case M_群发:
				if (!MyFun.isEmpty(_topic)) {
					_channel.basicPublish(_topic, "", null, msg);
				}
				break;
			case M_群发精确主题:
				if (!MyFun.isEmpty(_key) && !MyFun.isEmpty(_topic))
					_channel.basicPublish(_topic, _key, null, msg);
				break;
			case M_群发模糊主题:
				if (!MyFun.isEmpty(_key) && !MyFun.isEmpty(_topic))
					_channel.basicPublish(_topic, _key, null, msg);
				break;
			}
			if (_callback != null)
				_callback.success(m);
		} catch (Exception e) {
			if (_callback == null) {
				e.printStackTrace();
			} else {
				_callback.error(m, e);
			}
		}
	}

	String beforeReceive() throws Exception {
		String queueName = null;
		onBeforeAction();
		switch (_方式) {
		case M_一对一:
			return _key;
		case M_轮发:
			_channel.basicQos(1);
			return _key;
		case M_群发:
			queueName = _channel.queueDeclare().getQueue();
			_channel.queueBind(queueName, _topic, "");
			return queueName;
		case M_群发精确主题:
			queueName = _channel.queueDeclare().getQueue();
			_channel.queueBind(queueName, _topic, _key);
			return queueName;
		case M_群发模糊主题:
			queueName = _channel.queueDeclare().getQueue();
			_channel.queueBind(queueName, _topic, _key);
			return queueName;
		}
		return queueName;
	}

	@Override
	public void receive() {
		try {
			String queue = beforeReceive();
			if (_方式 == M_轮发)
				_is自动应答 = false;
			// 一对一模式无须手动应答,即使来不及处理,消息也会保留在服务器
			RbConsumer consumer = new RbConsumer(_channel, _callback, _is自动应答);
			_channel.basicConsume(queue, _is自动应答, consumer);
		} catch (Exception e) {
			if (_callback == null) {
				Log.error("接收消息异常--", e);
			} else {
				_callback.error(null, e);
			}
		}
	}

	public void removeGroup() {
		if (!MyFun.isEmpty(_topic))
			try {
				_channel.exchangeDelete(_topic);
			} catch (IOException e) {
				e.printStackTrace();
			}
	}

	public void remove() {
		if (!MyFun.isEmpty(_key))
			try {
				_channel.queueDelete(_key);
			} catch (IOException e) {
				e.printStackTrace();
			}
	}

	public void close() {
		// remove();
		// removeGroup();
		try {
			_channel.close();
		} catch (TimeoutException | IOException e) {
			e.printStackTrace();
		}
		try {
			connection.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}

class RbConsumer extends DefaultConsumer {
	IFinishT<CMsg> _接收处理 = null;
	// ISendReceive _isr=null;
	boolean _是否自动应答 = true;

	public RbConsumer(Channel channel, IFinishT<CMsg> domap, boolean 是否自动应答) {//
		super(channel);
		_接收处理 = domap;
		_是否自动应答 = 是否自动应答;
		// _isr=isr;
	}

	// String routingKey = ; // 队列名称
	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
			throws IOException {
		String title = envelope.getRoutingKey();
		CMsg result = new CMsg();
		result.s主题 = title;// 队列名称
		result._m属性 = properties.getHeaders();// 内容类型
		result.bs内容 = body;// 内容类型
		// System.out.println("应答标志"+envelope.getDeliveryTag());
		if (_接收处理 != null) {
			try {
				boolean 完成 = _接收处理.success(result);
				// 如果是手动应答,需要在事情已经处理完成后提交应答
				if (!_是否自动应答 && 完成)
					this.getChannel().basicAck(envelope.getDeliveryTag(), false);
			} catch (Exception e) {
				_接收处理.error(result, e);
				e.printStackTrace();
			}
		}
		/*
		 * System.out.println("消息：" +
		 * String.format("routingKey:%s;contentType:%s;consumerTag:%s;content:%s",
		 * routingKey, contentType, consumerTag,content));
		 */
		// 结束接收时调用应答
		// channel.basicAck(envelope.getDeliveryTag(), true);
		// 手动确认消息【参数说明：参数一：该消息的index；
		// 参数二：是否批量应答，true批量确认小于index的消息】
	}
}